package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

// 定义生产者和消费者函数类型
type ProducerFunc func(ctx context.Context, orderCh chan<- map[string]interface{}, errCh chan<- error) error
type ConsumerFunc func(ctx context.Context, orderCh <-chan map[string]interface{}, errCh chan<- error) error

// 模拟订单结构
type Order struct {
	ID     string  `json:"id"`
	Amount float64 `json:"amount"`
	Items  []Item  `json:"items"`
}

type Item struct {
	ProductID string `json:"product_id"`
	Quantity  int    `json:"quantity"`
}

// 生产者：模拟生成订单数据
func MockOrderProducer(ctx context.Context, orderCh chan<- map[string]interface{}, errCh chan<- error) error {
	defer log.Println("Producer stopped")

	orderID := 0
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			// 模拟生成订单
			orderID++
			order := map[string]interface{}{
				"id":     fmt.Sprintf("ORDER-%d", orderID),
				"amount": float64(orderID * 100),
				"items": []Item{
					{ProductID: "P1001", Quantity: 2},
					{ProductID: "P1002", Quantity: 1},
				},
			}

			// 发送到订单通道
			select {
			case orderCh <- order:
				log.Printf("Produced order: %s\n", order["id"])
			case <-ctx.Done():
				return nil
			}

			// 模拟生产间隔
			time.Sleep(1 * time.Second)
		}
	}
}

// 消费者：处理订单数据
func OrderProcessor(ctx context.Context, orderCh <-chan map[string]interface{}, errCh chan<- error) error {
	defer log.Println("Consumer stopped")

	for {
		select {
		case order, ok := <-orderCh:
			if !ok {
				return nil // 通道已关闭
			}

			// 将 map 转换为 Order 结构体
			data, err := json.Marshal(order)
			if err != nil {
				errCh <- fmt.Errorf("marshal error: %v", err)
				continue
			}

			var parsedOrder Order
			if err := json.Unmarshal(data, &parsedOrder); err != nil {
				errCh <- fmt.Errorf("parse error: %v", err)
				continue
			}

			// 模拟业务处理
			log.Printf("Processing order: %s (%.2f USD)\n",
				parsedOrder.ID, parsedOrder.Amount)
			time.Sleep(500 * time.Millisecond) // 模拟处理时间

		case <-ctx.Done():
			return nil
		}
	}
}

// 启动工作协程
func RunWorker(
	ctx context.Context,
	produce ProducerFunc,
	consume ConsumerFunc,
) (func(), error) {
	orderCh := make(chan map[string]interface{}, 10) // 带缓冲的订单通道
	errCh := make(chan error, 10)

	var wg sync.WaitGroup

	// 启动生产者
	wg.Add(1)
	go func() {
		defer wg.Done()
		if err := produce(ctx, orderCh, errCh); err != nil {
			errCh <- fmt.Errorf("producer error: %v", err)
		}
	}()

	// 启动消费者
	wg.Add(1)
	go func() {
		defer wg.Done()
		if err := consume(ctx, orderCh, errCh); err != nil {
			errCh <- fmt.Errorf("consumer error: %v", err)
		}
	}()

	// 错误处理协程
	wg.Add(1)
	go func() {
		defer wg.Done()
		for err := range errCh {
			log.Printf("ERROR: %v\n", err)
		}
	}()

	// 返回清理函数
	cleanup := func() {
		close(orderCh)
		close(errCh)
		wg.Wait()
	}

	return cleanup, nil
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 启动工作器
	cleanup, err := RunWorker(ctx, MockOrderProducer, OrderProcessor)
	if err != nil {
		log.Fatalf("Failed to start worker: %v", err)
	}

	// 捕获系统信号
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	// 等待信号
	<-sigCh
	log.Println("\nReceived shutdown signal")

	// 优雅关闭
	cancel()
	cleanup()
	log.Println("Service stopped gracefully")
}
